package com.g.localmsg.starter.service.impl;

import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.g.localmsg.starter.mapper.TransactionMapper;
import com.g.localmsg.starter.model.InvokeCtx;
import com.g.localmsg.starter.model.TaskState;
import com.g.localmsg.starter.model.TransactionMsgDO;
import com.g.localmsg.starter.service.TransactionService;
import com.g.localmsg.starter.util.ReflectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

@Service
public class TransactionServiceImpl implements TransactionService, ApplicationContextAware {

    private static final Logger log = LoggerFactory.getLogger(TransactionServiceImpl.class);

    @Autowired
    private TransactionMapper transactionMapper;
    @Autowired
    private ThreadPoolExecutor transactionExecutor;

    private ApplicationContext applicationContext;
    /**
     * 需要保证在事务中插入本地消息表，
     * @param transaction
     * @param async
     * @return
     */
    @Override
    public void executeDisTransaction(TransactionMsgDO transaction, boolean async) {
        if(!TransactionSynchronizationManager.isActualTransactionActive()){
            throw new RuntimeException("not in transaction");
        }
        transactionMapper.insert(transaction);
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                execute(transaction,async);
            }
        });
    }

    @Override
    public void processTransaction(TransactionMsgDO transaction) {
        String invokeCtxJson = transaction.getInvokeCtx();
        if (!StringUtils.hasText(invokeCtxJson)){
            String errMsg = "事务记录缺少调用上下文信息";
            log.error(errMsg);
            cancel(transaction, errMsg);
            return;
        }
        InvokeCtx invokeCtx = JSON.parseObject(invokeCtxJson, InvokeCtx.class);
        String className = invokeCtx.getClassName();
        String methodName = invokeCtx.getMethodName();
        Object[] args = invokeCtx.getArgs();
        List<Class<?>> paramTypes = ReflectionUtil.getParameterTypes(invokeCtx.getParameterTypes());
        try{
            transaction.setState(TaskState.PROCESSING.getCode());
            transactionMapper.updateById(transaction);
            Class<?> aClass = Class.forName(className);
            Method method = ReflectionUtil.getMethod(aClass, methodName, paramTypes);
            Object source = ReflectionUtil.getSourceObject(applicationContext.getBean(aClass));
            method.invoke(source, args);
            success(transaction);
        }catch (ClassNotFoundException e){

        } catch (Exception e){
            log.error("调用方法失败"+e);
            String failReason = e.getMessage();
            if (!StringUtils.hasLength(failReason)) {
                failReason = e.getCause() == null ? "未知错误" : e.getCause().getMessage();
            }
            retry(transaction, failReason);
        }
    }

    private void retry(TransactionMsgDO transaction, String failReason) {
        int currentRetryCount = transaction.getRetryCount()+ 1;
        transaction.setRetryCount(currentRetryCount);
        transaction.setFailedReason(failReason);
        if (currentRetryCount >= transaction.getMaxRetryCount()){
            transaction.setState(TaskState.CANCELED.getCode());
        }else {
            transaction.setNextRetryTime(LocalDateTime.now().plusSeconds(transaction.getRetryInterval()));
            transaction.setState(TaskState.FAILURE.getCode());
        }
        transactionMapper.updateById(transaction);
    }

    private void cancel(TransactionMsgDO transaction, String failReason) {
        transaction.setState(TaskState.CANCELED.getCode());
        transaction.setFailedReason(failReason);
        transaction.setRetryCount(transaction.getRetryCount() + 1);
        transactionMapper.updateById(transaction);
    }
    private void success(TransactionMsgDO transaction) {
        transaction.setState(TaskState.SUCCESS.getCode());
        transaction.setRetryCount(transaction.getRetryCount() + 1);
        transactionMapper.updateById(transaction);
    }
    @Override
    public void retryFailedTransaction(int batchSize) {
        QueryWrapper<TransactionMsgDO> wrapper = new QueryWrapper();
        wrapper.eq("state", TaskState.FAILURE.getCode()).or()
                .eq("state", TaskState.PENDING.getCode()).orderByAsc("next_retry_time");
        List<TransactionMsgDO> transactions = transactionMapper.selectList(wrapper);
        transactions.forEach(transaction -> execute(transaction,true));
    }

    private void execute(TransactionMsgDO transaction,boolean async){
        TransactionServiceImpl transactionService = currentProxy();
        if (async){
            transactionExecutor.execute(() -> {
                transactionService.processTransaction(transaction);
            });
        }else {
            transactionService.processTransaction(transaction);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
    private TransactionServiceImpl currentProxy() {
        return applicationContext.getBean(this.getClass());
    }
}
